package org.example

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object sparkBaseYun1 {
   def main(args: Array[String]): Unit = {
      val spark = SparkSession
        .builder()
        .master("local[*]")
        .appName("sparkBase")
        .getOrCreate()
      val sc = spark.sparkContext
//      不分区，数据就会分布式存储(分区数为电脑的最大核数）
      val rdd : RDD[Int] = sc.makeRDD(List(1,4,3,2,5),1)
//      map映射 filter过滤 sortBy排序
      val mapRDD = rdd.map(_ + 2).sortBy(tp => tp).filter(_ %2 == 0).take(1)
//      mapRDD.foreach(System.out.println)
//      将薪资top3的员工信息打印出来
//      val first_half = sc.textFile("C:\\Employee_salary_first_half")

//     1.创建RDD value value
      val rdd1 = sc.parallelize(List(1,2,3,4),2)
      val rdd2 = sc.parallelize(List(3,4,5,6),2)
//     交集 intersection
     val interRDD = rdd1.intersection(rdd2)
//     interRDD.foreach(System.out.println)
//     并集 union
     val unRDD = rdd1.union(rdd2)
//     unRDD.collect().foreach(System.out.println)
//     差集 站在rdd1的角度上，差集是 1 2
     val suRDD = rdd2.subtract(rdd1)
//     suRDD.foreach(System.out.println)
//     过滤 filter
     val score = sc.parallelize(List(('a',60),('b',70),('c',80),('a',60)))
//     score.filter(_._2 >= 80).collect().foreach(println)
//     去重 distinct
//     score.distinct().collect().foreach(println)
//     输出上半年或下半年实际薪资大于20万的员工姓名
// 从C盘根目录读取第一个CSV文件
val first_half = sc.textFile("C:\\Users\\Administrator\\Desktop\\Employee_salary_first_half.csv")
     // 从C盘根目录读取第二个CSV文件
     val second_half = sc.textFile("C:\\Users\\Administrator\\Desktop\\Employee_salary_second_half.csv")
     // 使用mapPartitionsWithIndex方法跳过CSV文件的标题行
     val drop_first = first_half.mapPartitionsWithIndex((ix,it) => {
       if (ix ==0) it.drop(1)
       it
     })
     val drop_second = second_half.mapPartitionsWithIndex((ix, it) => {
       if (ix == 0) it.drop(1)
       else it
     })
     // 将1中的每一行转换为(员工名, 工资)元组
     val split_first = drop_first.map(
       Line => {
         val data = Line.split(",");
         (data(1),data(6).toInt)
       })//使用逗号分割每行数据， 提取第二列和第七列数据，并将第七列转换为整数
     val split_second = drop_second.map(
       Line => {val data = Line.split(",");(data(1),data(6).toInt)})
     val filter_first=split_first.filter(x => x._2 > 200000.0).map(x => x._1)// 找出1中工资超过200,000的元组，并只保留员工名
     val filter_second=split_second.filter(x => x._2 > 200000.0).map(x => x._1)//x._n,n即使你要找的元素，通过 ._1 来访问第一个元素 a，通过 ._2 来访问第二个元素 b。
     val name=filter_first.union(filter_second).distinct()//合并并降重
//     name.collect().foreach(println)//逐行打印

//     键值对(key value) 元组 ._1 key ._2 value
      val data1 = sc.parallelize(List(('a',60),('b',70),('c',80),('a',60)))
      val data2 = sc.makeRDD(List(('a',6),('b',7),('c',8),('d',5)))
//      聚合 reduceByKey 相同的key做聚合
//     data1.reduceByKey((x,y) => x+y).foreach(println)
//     分组 groupByKey 相同的key分为一组
//      data1.groupByKey().foreach(println)
//      data1.groupBy(_._1).foreach(println)
//     连接 join 相同的key会依次匹配，没匹配上的数据就不会出现在结果上
//       data1.join(data2).foreach(println)
//     以左边为主，右边可能匹配不到 none
//       data1.leftOuterJoin(data2).foreach(println)
//     data1.rightOuterJoin(data2).foreach(println)
//     联合 适用于需要对相同key值进行聚合
       data1.combineByKey(
         v => (v, 1), //将key转为元祖
         (t:(Int, Int),v) => {
           (t._1+v,t._2+1)
         },
         (t1:(Int, Int),t2:(Int,Int)) => {
           (t1._1+t2._1, t1._2+t2._2)
         }
       ).foreach(println)
//     集合 求最小 最大值
     data1.aggregateByKey(0)(math.max(_,_),_+_).foreach(println)



      sc.stop()
   }

}
